Repository navigation
[runtime] Fix ActionStateStore divergence detection for multi-action replay - #1306
Merged
wenjin272 merged 1 commit intoOct 22, 2026
Merged
Conversation
…replay Replace action-count divergence checks with indexed input fingerprints and completed-producer event evidence shared by Kafka and Fluss. Preserve normal fan-out and cascades while invalidating conflicting histories and stale child states. Keep recovery offsets consistent and refresh Fluss records in sequence order. Add index lifecycle, repeated recovery, real backend and Python bridge regression coverage. Document shared Memory ordering limits during replay. Fixes apache#1301 Validation: Java 11 regression and native Python suites: 384 passed. Java 21 expanded regression: 394 passed, 14 skipped by existing continuation assumptions. Spotless, Ruff, license checks and git diff --check passed. Generated-by: Codex 0.153.4 (GPT-6) Co-authored-by: Codex <codex@openai.com>
wenjin272
added
bug
github-actions
Bot
added
doc-included
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Linked issue: #1301
Fixes #1301
Purpose of change
Normal fan-out and cascades now retain later saved Action completions during recovery. Previously, both stores treated multiple ActionStates for one key/sequence as divergence, causing later inputs to rerun unnecessarily.
Runtime flow
Kafka and Fluss rebuild a shared indexed cache incrementally from their ordered logs. Each lookup checks input fingerprints and, where available, completed producers' output Event IDs before returning a saved state. Conflicts invalidate the affected history; ordinary misses retain the existing conservative suffix invalidation. Kafka removes corresponding recovery offsets. Fluss refreshes cached rounds in sequence order before checkpointing so a subsequent rebuild preserves recovered future work.
Key decisions
Action counts do not identify a fork. Root matching keeps the existing attribute fingerprint because restored InputEvents can have new random IDs. Internal occurrence checks require persisted completed-producer evidence; comparing every random Event ID would incorrectly reject regenerated subgraph bootstraps. A shared per-key, ordered-sequence index avoids whole-cache scans in lookup and log replay.
Behavioral Semantics
Interaction decisions
All invalidation below is restricted to the same business key.
Behavioral contracts
Failure behavior
Detected conflicts and misses fall back to execution rather than failing recovery. The existing suffix invalidation remains conservative even when a miss does not prove divergence. Backend write/rebuild failures retain the stores' existing exception behavior; failed writes do not add new indexed producer evidence. No new configuration or backend schema is introduced.
Tests
ActionStateDivergenceRecoveryTestnormal-history cases;ActionStatePythonDivergenceITActionStateDivergenceRecoveryTestchanged-input, changed-root and truncated-replay cases;ActionStateDivergenceIntegrationTest.changedRootCannotReuseChildComputedAgainstAnotherMemoryHistoryActionStateDivergenceRecoveryTest.checkpointAfterDivergenceMustNotReviveOldSuffixand repeated-restore casesActionStateCacheTestcross-key/boundary cases; regenerated-bootstrap recovery caseKafkaActionStateStoreTest.divergenceRemovesOffsetsFromRecoveryMarkers;ActionStateDivergenceIntegrationTest.completedCascadeSurvivesCheckpointBeforeReplayingRecoveredInputsJava 11: 380 regression tests plus 4 native Python/Pemja tests passed, with no skips. Java 21 clean rebuild: 394 passed, 14 skipped by existing non-continuation assumptions in
ActionExecutionOperatorTest; those cases passed on Java 11. All new tests executed on both JDKs. The integration suite used an actual Kafka container and embedded Fluss cluster. Spotless (Java 11), Ruff, license checks and whitespace checks passed.Not verified: the full repository suite, exhaustive retention/rescaling combinations, a throughput or heap benchmark, and a dedicated shared-Memory interleaving reproduction. Python/Pemja tests use simulated backend transport; live backend tests use Java Actions. This does not solve cross-channel input ordering (#1302), general same-attribute occurrence collisions (#1084), or arbitrary nondeterminism.
Index invariants and reproducible test selections
Each key has an ordered sequence map with state keys, input fingerprint counts and completed-output ID counts. Per-state evidence snapshots handle ActionState objects mutated in place before put. Put/remove/clear, tombstones and pruning maintain these indexes; map views cannot bypass maintenance.
Lookup costs O(log R), indexing O(log R + E), and invalidation visits affected rounds/states. R is cached rounds for one key; E is the record's output count. Fluss checkpoint refresh still enumerates surviving states. The NoScanMap regression prohibits whole-map traversal over 3,000 states across 100 keys. Nine cache tests cover evidence reference counts, mutation, deletion and sequence bounds.
Run from the checkout with the selected JDK and a supported Python on PATH. Set PYTHONPATH to the checkout's python directory, runtime/src/test/resources and the active Python's purelib directory. Local Mockito runs used a Byte Buddy 1.15.4 startup agent.
This is the Java 21 expanded selection. Java 11 ran the state-store suites, DurableExecutionManagerTest, ActionExecutionOperatorTest and MemoryWriteBelowValueRecoveryTest, plus the Python suite separately. The native Python test ends in IT and must be selected explicitly. Docker is required for Kafka integration coverage.
API
No public API, configuration, persisted key format or payload schema changes. Existing stores gain corrected cache invalidation and reuse behavior. Memory overhead increases for the indexes; recovery still loads the required log window into memory.
Documentation
doc-neededdoc-not-neededdoc-includedConsistency now documents shared Memory ordering under Known limitations, alongside external side effects. The Actions chapter links to the combined section.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex 0.153.4 (GPT-6)